package com.fengkai.filepost.pdfoutstream.test.common.mq.translation;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Fengkai
 * @creat 2021/12/30
 */
public class TransactionProducer {
    //定义事务消息生产者
       public static void main(String[] args) throws MQClientException {
           TransactionMQProducer transactionMQProducer = new TransactionMQProducer("tpg");
           transactionMQProducer.setNamesrvAddr("192.168.157.129:9876");

           //定义一个线程池
           ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                   new ArrayBlockingQueue<>(200),
                   new ThreadFactory() {
                       @Override
                       public Thread newThread (Runnable r) {
                           Thread thread = new Thread();
                           thread.setName("transactionProducer-check-thread");
                           return thread;
                       }
                   });

           //为生产者指定一个线程池
           transactionMQProducer.setExecutorService(threadPoolExecutor);

           //指定监听器
           transactionMQProducer.setTransactionListener(new ICBCTransactionListener());

           transactionMQProducer.start();
           String[] tags = {"TagA","TagB","TagC"};
           for (int i = 0; i < 3; i++) {
               byte[] body = ("Hi," + i).getBytes();
               Message msg = new Message("TranscationTopic", tags[i], body);
               // 发送事务消息
               // 第二个参数用于指定在执行本地事务时要使用的业务参数
               SendResult sendResult = transactionMQProducer.sendMessageInTransaction(msg,null);
               System.out.println("发送结果为：" + sendResult.getSendStatus());
           }
           transactionMQProducer.shutdown();
       }

}
